import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;
import java.io.*;
//import java.util.Arrays;
import java.util.Arrays;
//import java.util.System;
//import java.util._;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.BufferedReader;
import java.util.Iterator;


import java.lang.Iterable;


public class wordcount {
    public static void main(String[] args)
    {
        SparkConf conf = new SparkConf().setMaster("local").setAppName("wordcount1");
        JavaSparkContext sc = new JavaSparkContext(conf);

        String filename = "/home/appleyuchi/桌面/spark_success/wordcount/Java/src/main/java/hello.txt";
        JavaRDD<String> input = sc.textFile(filename);

        JavaRDD<String> lines = input.flatMap(new FlatMapFunction<String, String>()
        {

            public Iterator<String> call(String s) throws Exception

            {
                return Arrays.asList(s.split(" ")).iterator();
            }
        }
        );






        //pairs
        JavaPairRDD<String,Integer> pairs = lines.mapToPair(new PairFunction<String, String, Integer>()
        {
            public Tuple2<String, Integer> call(String s) throws Exception
            {
                return new Tuple2<String, Integer>(s,1);
            }
        });

        //reduce
        JavaPairRDD<String,Integer> counts = pairs.reduceByKey(new Function2<Integer, Integer, Integer>()
        {
            public Integer call(Integer x, Integer y) throws Exception
            {
                return x+y;
            }
        });

        //output
        counts.foreach(new VoidFunction<Tuple2<String, Integer>>()
        {
            public void call(Tuple2<String, Integer> tuple2) throws Exception
            {
                System.out.println(tuple2);
            }
        });

        sc.stop();

    }
}